package com.maben.mqtt.test002;

import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

import java.nio.charset.StandardCharsets;

/**
 * 简单订阅端
 */
public class Subscribe {
    public static void main(String[] args) throws Exception{
        final String broker="tcp://broker.emqx.io:1883";
        final String clientId = "subscribe_client";
        final String topic = "mqtt/test";
        final String username = "emqx";
        final String password = "public";
        /*
            QoS 0：消息最多传递一次，如果当时客户端不可用，则会丢失该消息。
            QoS 1：消息传递至少 1 次。
            QoS 2：消息仅传送一次。
         */
        int qos=0;

        //设置mqtt客户端
        final MqttClient client = new MqttClient(broker, clientId, new MemoryPersistence());
        //连接参数
        MqttConnectOptions options = new MqttConnectOptions();
        options.setUserName(username);
        options.setPassword(password.toCharArray());
        options.setKeepAliveInterval(60);
        /*
            一般情况下 MQTT 客户端仅能接收到在线时其他客户端发布的消息，
            如果客户端离线再上线后将收不到离线期间的消息。
            但是当客户端使用固定的 Client ID，且连接参数 Clean Session 为 false 时，
            客户端离线后消息服务器可以为客户端保持一定量的离线消息，
            并在客户端再次上线后发送给客户端（且为客户端恢复下线前的订阅信息）。
            需要配合使用：cleanSession为false，且 QoS 设置为 1。
         */
        options.setCleanSession(true);
        //【客户端】设置回调函数
        client.setCallback(new MqttCallback() {
            /**
             * 连接丢失时被调用
             */
            public void connectionLost(Throwable throwable) {
                System.out.println("connectionLost: " + throwable.getMessage());
            }

            /**
             * 接收到消息时被调用
             */
            public void messageArrived(String topic, MqttMessage message) throws Exception {
                System.out.println("topic: " + topic);
                System.out.println("Qos: " + message.getQos());
                System.out.println("message content: " + new String(message.getPayload(), StandardCharsets.UTF_8));
            }

            /**
             * 消息发送完成时被调用
             */
            public void deliveryComplete(IMqttDeliveryToken token) {
                System.out.println("deliveryComplete---------" + token.isComplete());
            }
        });
        //连接
        client.connect(options);
        //订阅
        client.subscribe(topic,qos);
        System.out.println("******订阅成功*******");
    }
}
